﻿#include <tcp_server.h>
#include <QTimer>
tcp_server::tcp_server(int Thread_Cout, int port)
{
    qRegisterMetaType<emit_type>("emit_type");
    this->Thread_Cout=Thread_Cout;
    this->port=port;
    for(int cout=0;cout<Thread_Cout;cout++)
    {
        QThread *thread=new QThread();
        Thread_Socket_Cout[thread]=0;
        thread->start();
    }
    TcpClock.start(30*1000);
}

tcp_server::~tcp_server()
{
    qDebug()<<"tcp_server析构";
    emit disconnect_Allsocket();
    QMap  <tcp_socket*,QThread*>::iterator i;
    for (i = Socket_Thread.begin(); i != Socket_Thread.end(); ++i) {    //对QMAP进行遍历
        i.value()->quit();
        i.value()->wait();
        i.value()->deleteLater();

    }
}

bool tcp_server::Start_Server()
{
    if (!listen(QHostAddress::Any, port)) {
        qDebug() << "监听失败!!";
        return false;
    }
    else {
        qDebug() << "开始监听,端口号为:"<<port;
        return true;
    }
}

QThread* tcp_server::Choice_Free_Thread()                                          //选择负载最小的线程
{
    int thread_min_cout=INT_MAX;
    QThread *thread=NULL;
    QMap  <QThread*,int>::const_iterator i;
    for (i = Thread_Socket_Cout.constBegin(); i != Thread_Socket_Cout.constEnd(); ++i) {    //对QMAP进行遍历
        if(i.value()<thread_min_cout)//选择负载最小线程
        {
            thread_min_cout=i.value();
            thread=i.key();
        }
    }
    return thread;
}

void tcp_server::incomingConnection(qintptr socketDescriptor)                     //当有新用户接入时触发此程序
{
    qDebug()<<"新用户连接";
    QThread *thread=Choice_Free_Thread();      //选择负载最小的线程
    if(thread!=NULL)
    {
        tcp_socket *socket=new tcp_socket(socketDescriptor);
        if(!socket->tcpSocket_init())
        {
            socket->deleteLater();
            return;
        }
        socket->send_message("服务器","获取程序类型","");
        QSharedPointer<M_Socket>   msocket=QSharedPointer<M_Socket>(new M_Socket());
        msocket.data()->Stype=Tsocket;
        msocket.data()->t_socket=socket;
        Common::tcp_server_SocketMapLock.lockForWrite();
        Common::tcp_server_SocketMap[socket]=msocket;                //tcp_server_SocketMap添加socket
        Common::tcp_server_SocketMapLock.unlock();
        Socket_Thread[socket]=thread;                                //记录socket所在线程
        Thread_Socket_Cout[thread]+=1;                               //增添线程socket数量
        connect(this,SIGNAL(disconnect_Allsocket()),socket,SLOT(disconnect_Allsocket()));
        connect(socket,SIGNAL(disconnected()),this,SLOT(delete_tcpSocket()),Qt::QueuedConnection);
        connect(this,SIGNAL(delete_socket(tcp_socket*)),socket,SLOT(socket_delete(tcp_socket*)),Qt::QueuedConnection);
        connect(socket,SIGNAL(send_ReplyData(QString,QString,QString)),handlemanage.data(),SLOT(receive(QString,QString,QString)),Qt::QueuedConnection);
        socket->moveToThread(thread);
        connect(handlemanage.data(),SIGNAL(t_SocketSendData(QString,QString,QString,tcp_socket*,emit_type,QString)),socket,SLOT(send_messagefromslot(QString,QString,QString,tcp_socket*,emit_type,QString)),Qt::QueuedConnection);
        connect(&TcpClock,SIGNAL(timeout()),socket,SLOT(IsConnect()),Qt::QueuedConnection);
    }
    else
    {
        qDebug()<<"目前所有线程达到警戒值，拒绝接入!";
    }
}

void tcp_server::delete_tcpSocket()
{
    QStringList taskid;
    QMap  <tcp_socket*,QThread*>::iterator i;
    tcp_socket *socket=qobject_cast<tcp_socket *>(sender());
    Common::tcp_server_SocketMapLock.lockForWrite();
    Thread_Socket_Cout[Socket_Thread[Common::tcp_server_SocketMap[socket].data()->t_socket]]-=1;                  //线程对应socket-1
    taskid=Common::tcp_server_SocketMap[socket].data()->task_id;
    if(Socket_Thread.remove(Common::tcp_server_SocketMap[socket].data()->t_socket)==0)                            //移除
    {
        qDebug()<<"删除socket_Thread失败!";
    }
    if(Common::tcp_server_SocketMap.remove(socket)==0)
    {
        qDebug()<<"删除socket时,tcp_server_SocketMap列表没有找到此socket,可能造成内存泄漏";
    }
    Common::tcp_server_SocketMapLock.unlock();
    Common::CaculateTask_lock.lockForRead();
    for(int a=0;a<taskid.size();a++)
    {
        QString belongid=taskid[a].split("-").first();
        if(!Common::CaculateTask.contains(belongid))
        {
            qDebug()<<"未找到相应任务!";
            continue;
        }
        qDebug()<<belongid<<taskid[a];
        Common::CaculateTask[belongid].data()->Caculating_task_lock.lockForWrite();
        if(!Common::CaculateTask[belongid].data()->Decomposition_Caculating_task.contains(taskid[a]))
        {
            qDebug()<<"不存在相应任务";
            Common::CaculateTask[belongid].data()->Caculating_task_lock.unlock();
            return;
        }
        Common::CaculateTask[belongid].data()->Decomposition_Caculating_task[taskid[a]].data()->DTask_Lock.lockForWrite();
        Common::CaculateTask[belongid].data()->Decomposition_Caculating_task[taskid[a]].data()->setProblemStatus(task_free);
        Common::CaculateTask[belongid].data()->Decomposition_Caculating_task[taskid[a]].data()->DTask_Lock.unlock();
        QSharedPointer<Decomposition_Task> D_task= Common::CaculateTask[belongid].data()->Decomposition_Caculating_task[taskid[a]];
        if(Common::CaculateTask[belongid].data()->Decomposition_Caculating_task.remove(taskid[a])==0)
        {
            qDebug()<<"计算节点断开时,移除任务失败!";
        }
        Common::CaculateTask[belongid].data()->Caculating_task_lock.unlock();
        Common::CaculateTask[belongid].data()->unfinish_task_lock.lockForWrite();
        Common::CaculateTask[belongid].data()->Decomposition_unfinish_task[taskid[a]]=D_task;
        Common::CaculateTask[belongid].data()->unfinish_task_lock.unlock();

    }
    Common::CaculateTask_lock.unlock();
    emit delete_socket(socket);
}


void tcp_server::set_Handle(QWeakPointer<handle_manage> handlemanage)
{
    this->handlemanage=handlemanage;
}
